package cn.mesmile.flink.watermark;

import cn.hutool.core.date.DateUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * @author zb
 * @date 2021/8/25 23:20
 * @Description
 */
public class FlinkWatermarkSideOutputLatenessAPP {

    /**
     * 窗口 [23:12:00 ~ 23:12:10) | [23:12:10 ~ 23: 12:20)
     * 触发窗口计算条件
     * 窗口内有数据
     * watermark >= 窗口endtime
     * 即 当前计算窗口最大的事件时间 - 允许乱序延迟的时间 >= Window EndTime窗口结束时间
     * java,2022-11-11 23:12:07,10
     * java,2022-11-11 23:12:11,10
     * java,2022-11-11 23:12:08,10
     * java,2022-11-11 23:12:13,10
     * java,2022-11-11 23:12:23,10
     * #延迟1分钟内，所以会输出
     * java,2022-11-11 23:12:09,10
     * java,2022-11-11 23:12:02,10
     * java,2022-11-11 23:14:30,10
     * #延迟超过1分钟，不会输出，配置了sideOutPut，会在兜底输出
     * java,2022-11-11 23:12:03,10
     * java,2022-11-11 23:12:04,10
     * @param args
     * @throws Exception
     */

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // java,2021-11-11 09:30:33,20
        DataStream<String> ds = env.socketTextStream("127.0.0.1",8888);

        DataStream<Tuple3<String, String,Integer>> flatMapDS = ds.flatMap(new FlatMapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple3<String, String,Integer>> out) throws Exception {
                String[] arr = value.split(",");
                out.collect(Tuple3.of(arr[0], arr[1],Integer.parseInt(arr[2])));
            }
        });

        SingleOutputStreamOperator<Tuple3<String, String,Integer>> watermakerDS = flatMapDS
                .assignTimestampsAndWatermarks(WatermarkStrategy
                //指定最大允许的延迟/乱序 时间 这里指定为 3秒
                .<Tuple3<String, String,Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                        (event, timestamp) -> {
                            //指定POJO的事件时间列，返回毫秒
                            return DateUtil.parseDateTime(event.f1).getTime();
                        }
                ));

        OutputTag<Tuple3<String, String,Integer>> lateData = new
                OutputTag<Tuple3<String, String,Integer>>("lateData"){};

        SingleOutputStreamOperator<String> sumDS = watermakerDS.keyBy(new KeySelector<Tuple3<String, String,Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String,Integer> value) throws Exception {
                return value.f0;
            }
        })// 开窗
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 允许 1分钟的延迟
                .allowedLateness(Time.minutes(1))
                // 最后一层兜底
                .sideOutputLateData(lateData)
                .apply(new WindowFunction<Tuple3<String, String,Integer>, String, String, TimeWindow>() {
            @Override
            public void apply(String key, TimeWindow window, Iterable<Tuple3<String, String,Integer>> input, Collector<String> out) throws Exception {
                //存放窗口的数据的事件时间
                List<String> eventTimeList = new ArrayList<>();
                int total = 0;
                for (Tuple3<String, String,Integer> order : input) {
                    eventTimeList.add(order.f1);
                    total = total+order.f2;
                }
                String outStr = String.format("分组key:%s,聚合值:%s,窗口开始结束:[%s~%s),窗口所有事件时间:%s", key,total, DateUtil.formatDateTime(new Date(window.getStart())),DateUtil.formatDateTime(new Date(window.getEnd())), eventTimeList);
                out.collect(outStr);
            }
        });


        //不会更新之前的窗口数据，需要代码单独写逻辑处理更新之前的数据，也可以积累后批处理
        sumDS.getSideOutput(lateData).print("late data");

        sumDS.print("result:");

        env.execute("watermark job");
    }
}
